package com.atguigu.flink.sql.function;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;

/**
 * Created by Smexy on 2023/4/14
 *
 *  Scalar functions map scalar values to a new scalar value.
 *          一行N列进，一行一列出的函数。
 *
 *          把字符串转换为大写
 *
 *   ①按照flink的规范，定义一个函数类
 *   ②在系统中注册函数，提供一个函数名字
 *   ③可以调用函数
 */
public class Demo3_CustomScalarFunciton
{
    public static void main(String[] args) {

        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().inStreamingMode().build();
        TableEnvironment tableEnvironment = TableEnvironment.create(environmentSettings);

        String createTableSql = " create table t1 ( id STRING, ts BIGINT , vc INT ," +
            "  `file.path` STRING NOT NULL METADATA " +
            "    )WITH (" +
            "  'connector' = 'filesystem'," +
            "  'path' = 'data/sensor.txt'," +
            "  'format' = 'csv'" +
            ") ";

        tableEnvironment.executeSql(createTableSql);

        //②在系统中注册函数，提供一个函数名字
        MyUpper myFunction = new MyUpper();
        tableEnvironment.createTemporaryFunction("myUpper",myFunction);

        tableEnvironment.sqlQuery(" select *,myUpper(id) randNum from t1  ").execute().print();

    }

    /*
        按照flink的规范，定义一个函数类

        想要实现自定义标量函数，你需要扩展 org.apache.flink.table.functions 里面的 ScalarFunction 并且实现一个或者多个求值方法。
        标量函数的行为取决于你写的求值方法。求值方法必须是 public 的，而且名字必须是 eval。
     */

    public static class MyUpper extends ScalarFunction{

        /*
            求值方法必须是 public 的，而且名字必须是 eval.
            返回值不能是void
         */
        public String eval(String s){
            if (s == ""){
                return "default";
            }else {
                return s.toUpperCase();
            }
        }
    }
}
